"""
Case Type   : Ustore_Standby_Logicreplication
Case Name   : 删除逻辑复制槽权限验证
Create At   : 2022/03/22
Owner       : zhao-pengcheng-g
Description :
    1.备份postgresql.conf,pg_hba.conf文件
    2.修改ustore参数
    3.修改逻辑复制相关参数
    4.创建用户,配置白名单
    5.在主机上创建复制槽
    6.开启复制槽
    7.在主机上删除运行中的复制槽
    8.停止主机运行中复制槽
    9.在主备节点上删除复制槽
    10.在主机上删除不存在的复制槽
    11.资源清理，恢复参数
Expect      :
    1.备份成功
    2.修改ustore参数成功
    3.修改逻辑复制相关参数成功
    4.创建成功，配置白名单成功
    5:创建成功
    6:开启成功
    7:删除在主机上运行中的复制槽失败，删除在备机上运行中的复制槽成功
    8.停止逻辑复制槽成功
    9.在备机上删除复制槽失败合理报错，在主机上删除成功
    10.合理报错
    11.资源清理成功，恢复参数成功
History     :
"""
import os
import time
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread

Primary_SH = CommonSH('PrimaryDbUser')

@unittest.skipIf(2 != Primary_SH.get_node_num(),'非1+1环境不执行')
class StandbyLogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.standby_node = Node('Standby1DbUser')
        self.sh_primy = CommonSH('PrimaryDbUser')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.dir_replication = os.path.join(macro.DB_BACKUP_PATH,
                                            'dir_replication_0059')
        self.postgresql_path = os.path.join(macro.DB_INSTANCE_PATH,
                                            macro.DB_PG_CONFIG_NAME)
        self.postgresql_bak_path = os.path.join(self.dir_replication,
                                                'postgresql_bak.conf')
        self.pg_hba_path = os.path.join(macro.DB_INSTANCE_PATH,
                                        macro.PG_HBA_FILE_NAME)
        self.pg_hba_bak_path = os.path.join(self.dir_replication,
                                            'pg_hba_bak.conf')
        self.db_name = 'db_replication_0059'
        self.user_name = 'u_logic_replication_0059'
        self.slot_name_1 = 'slot_standby_logic_replication_0059_1'
        self.slot_name_2 = 'slot_standby_logic_replication_0059_2'
        self.log_file_1 = os.path.join(self.dir_replication,
                                       'standby_logic_replication_0059_1.log')
        self.log_file_2 = os.path.join(self.dir_replication,
                                       'standby_logic_replication_0059_2.log')

        text = "-----step1:备份postgresql.conf,pg_hba.conf;expect:成功-----"
        self.logger.info(text)
        cp_cmd = f'''rm -rf {self.dir_replication} && 
                    mkdir {self.dir_replication} &&
                    cp {self.postgresql_path} {self.postgresql_bak_path} && 
                    cp {self.pg_hba_path} {self.pg_hba_bak_path} && echo OK'''
        cp_pri_msg = self.primary_node.sh(cp_cmd).result()
        self.logger.info(cp_pri_msg)
        self.assertEqual("OK", cp_pri_msg, "执行失败" + text)
        cp_sta_msg = self.standby_node.sh(cp_cmd).result()
        self.logger.info(cp_sta_msg)
        self.assertEqual("OK", cp_sta_msg, "执行失败" + text)

        text = "-----step2:修改ustore参数;expect:成功-----"
        self.logger.info(text)
        self.config_ustore = 'enable_default_ustore_table'
        self.show_ustore_cmd = f'show {self.config_ustore};'
        check_res = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(check_res)
        self.ustore_default = check_res.splitlines()[-2].strip()
        if 'on' != self.ustore_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_ustore}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_ustore, 'on',
                                  '(1 row)', flag='1')

        text = "-----step3:修改逻辑复制相关参数;expect:成功-----"
        self.logger.info(text)
        self.config_slot_log = 'enable_slot_log'
        self.show_slot_log_cmd = f'show {self.config_slot_log};'
        check_res = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(check_res)
        self.slot_log_default = check_res.splitlines()[-2].strip()
        if 'on' != self.slot_log_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_slot_log,
                                  'on', '(1 row)', flag='1')

        self.config_wal_level = 'wal_level'
        self.show_wal_level_cmd = f'show {self.config_wal_level};'
        check_res = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(check_res)
        self.wal_level_default = check_res.splitlines()[-2].strip()
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primy.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primy.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_wal_level,
                                  'logical', '(1 row)', flag='1')
        status = self.sh_primy.get_db_cluster_status()
        self.assertNotIn("ERROR", status, '执行失败:' + text)
        self.assertIn("Normal", status, '执行失败:' + text)

        text = "-----step4:创建用户，并设置权限和白名单;expect:成功-----"
        self.logger.info(text)
        create_cmd = f'''drop user if exists {self.user_name};
                    create user {self.user_name} sysadmin \
                    password '{macro.COMMON_PASSWD}';'''
        create_msg = self.sh_primy.execut_db_sql(create_cmd,
                                                 dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG,
                      create_msg, '执行失败' + text)
        param_1 = f'local replication {self.user_name} sha256'
        param_2 = f'host replication {self.user_name} 127.0.0.1/32 sha256'
        param_3 = f'host replication {self.user_name} ' \
            f'{self.primary_node.db_host}/32 sha256'
        result_1 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_1)
        result_2 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_2)
        result_3 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

    def test_standby_logicreplication(self):
        text = "-----step5:在主机上创建复制槽;expect:创建成功-----"
        self.logger.info(text)
        names = [self.slot_name_1, self.slot_name_2]
        for name in names:
            create_slot = f"pg_recvlogical -o skip-empty-xacts=true " \
                f"-d postgres -S {name} " \
                f"-p {self.primary_node.db_port} " \
                f"-U {self.user_name} -s 2 --create"
            execute_cmd = f'''source {macro.DB_ENV_PATH};
                        expect <<EOF
                        set timeout 300
                        spawn {create_slot}
                        expect "Password:"
                        send "{macro.COMMON_PASSWD}\\n"
                        expect eof\n''' + '''EOF'''
            self.logger.info(execute_cmd)
            execute_msg = self.primary_node.sh(execute_cmd).result()
            self.logger.info(execute_msg)
        text = "-----验证创建逻辑复制槽成功;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"select * from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1 and self.slot_name_2,
                      sql_msg, '执行失败:' + text)

        text = "-----step6:开启复制槽;expect:开启成功-----"
        self.logger.info(text)
        primy_start_cmd = f"pg_recvlogical " \
            f"-o skip-empty-xacts=true -F 1 " \
            f"-d postgres -S {self.slot_name_1} " \
            f"-p {self.primary_node.db_port} -s 2 " \
            f"-f {self.log_file_1} " \
            f"-U {self.user_name} " \
            f"--start "
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                        expect <<EOF 
                        set timeout 300 
                        spawn {primy_start_cmd} 
                        expect "Password:" 
                        send "{macro.COMMON_PASSWD}\\n" 
                        expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        thread_1 = ComThread(self.common.get_sh_result,
                             args=(self.primary_node, execute_cmd,))
        thread_1.setDaemon(True)
        thread_1.start()
        thread_1.join(10)
        msg_result_1 = thread_1.get_result()
        self.logger.info(msg_result_1)
        text = "-----验证开启逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select slot_name,active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_1}%';;"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1 and 't', sql_msg.splitlines()[-2],
                      '执行失败:' + text)

        stanby_1_start_cmd = f"pg_recvlogical " \
            f"-o skip-empty-xacts=true -F 1 " \
            f"-d postgres -S {self.slot_name_2} " \
            f"-p {self.standby_node.db_port} -s 2 " \
            f"-f {self.log_file_2} " \
            f"-U {self.user_name} " \
            f"--start "
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                        expect <<EOF 
                        set timeout 300 
                        spawn {stanby_1_start_cmd} 
                        expect "Password:" 
                        send "{macro.COMMON_PASSWD}\\n" 
                        expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        thread_2 = ComThread(self.common.get_sh_result,
                             args=(self.standby_node, execute_cmd,))
        thread_2.setDaemon(True)
        thread_2.start()
        thread_2.join(10)
        msg_result_2 = thread_2.get_result()
        self.logger.info(msg_result_2)
        text = "-----验证开启逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select slot_name,active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_2}%';;"
        sql_msg = self.sh_standby.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_2 and 't', sql_msg.splitlines()[-2],
                      '执行失败:' + text)

        text = "-----step7:在主机上删除运行中的复制槽;expect:删除在主机上运行" \
               "中的复制槽失败，删除在备机上运行中的复制槽成功-----"
        self.logger.info(text)
        names = [self.slot_name_1, self.slot_name_2]
        for name in names:
            drop_slot = f"pg_recvlogical -o skip-empty-xacts=true " \
                f"-F 1 -d postgres -S {name} " \
                f"-p {self.primary_node.db_port} " \
                f"-U {self.user_name} -s 2 --drop"
            execute_cmd = f'''source {macro.DB_ENV_PATH};
                        expect <<EOF
                        set timeout 300
                        spawn {drop_slot}
                        expect "Password:"
                        send "{macro.COMMON_PASSWD}\\n"
                        expect eof\n''' + '''EOF'''
            self.logger.info(execute_cmd)
            execute_msg = self.primary_node.sh(execute_cmd).result()
            self.logger.info(execute_msg)
            if name == self.slot_name_1:
                self.assertIn("already active", execute_msg, '执行失败:' + text)
            else:
                self.assertNotIn("ERROR", execute_msg, '执行失败:' + text)

        text = "-----step8:停止复制槽;expect:停止复制槽成功-----"
        self.logger.info(text)
        stop_cmd = "ps -ef | grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs kill -9"
        self.logger.info(stop_cmd)
        primy_result = self.primary_node.sh(stop_cmd).result()
        self.logger.info(primy_result)
        standy_result = self.standby_node.sh(stop_cmd).result()
        self.logger.info(standy_result)
        time.sleep(5)

        text = "-----step9:在主备节点上删除复制槽，" \
               "expect:在备机上删除复制槽失败合理报错，在主机上删除成功-----"
        self.logger.info(text)
        drop_slot = f"pg_recvlogical -o skip-empty-xacts=true " \
            f"-d postgres -S {self.slot_name_1} " \
            f"-p {self.standby_node.db_port} " \
            f"-U {self.user_name} -s 2 --drop"
        execute_cmd = f'''source {macro.DB_ENV_PATH};
                        expect <<EOF
                        set timeout 300
                        spawn {drop_slot}
                        expect "Password:"
                        send "{macro.COMMON_PASSWD}\\n"
                        expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        execute_msg = self.standby_node.sh(execute_cmd).result()
        self.logger.info(execute_msg)
        self.assertIn("could not send replication command",
                      execute_msg, '执行失败:' + text)

        text = "-----在主机上删除逻辑复制槽，expect:删除成功-----"
        self.logger.info(text)
        drop_slot = f"pg_recvlogical -o skip-empty-xacts=true " \
            f"-d postgres -S {self.slot_name_1} " \
            f"-p {self.primary_node.db_port} " \
            f"-U {self.user_name} -s 2 --drop"
        execute_cmd = f'''source {macro.DB_ENV_PATH};
                                    expect <<EOF
                                    set timeout 300
                                    spawn {drop_slot}
                                    expect "Password:"
                                    send "{macro.COMMON_PASSWD}\\n"
                                    expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        execute_msg = self.primary_node.sh(execute_cmd).result()
        self.logger.info(execute_msg)
        self.logger.info("-----验证是否删除成功;expect:成功-----")
        sql_cmd = f"select * from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertNotIn(self.slot_name_1 and self.slot_name_2,
                         sql_msg, '执行失败:' + text)

        text = "-----step10:删除不存在的复制槽，expect:合理报错-----"
        self.logger.info(text)
        names = [self.slot_name_1, self.slot_name_2]
        for name in names:
            drop_slot = f"pg_recvlogical -o skip-empty-xacts=true " \
                f"-d postgres -S {name} " \
                f"-p {self.primary_node.db_port} " \
                f"-U {self.user_name} -s 2 --drop"
            execute_cmd = f'''source {macro.DB_ENV_PATH};
                        expect <<EOF
                        set timeout 300
                        spawn {drop_slot}
                        expect "Password:"
                        send "{macro.COMMON_PASSWD}\\n"
                        expect eof\n''' + '''EOF'''
            self.logger.info(execute_cmd)
            execute_msg = self.primary_node.sh(execute_cmd).result()
            self.logger.info(execute_msg)
            self.assertIn("does not exist", execute_msg, '执行失败:' + text)
        
    def tearDown(self):
        text = "-----step11:资源清理，恢复参数，expect:成功-----"
        self.logger.info(text)
        text_1 = "-----step11.1:删除用户，expect:成功-----"
        self.logger.info(text_1)
        drop_cmd = f'''drop user {self.user_name};'''
        drop_msg = self.sh_primy.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)

        text_2 = "-----step11.2:恢复postgresql.conf,pg_hba.conf文件，" \
                 "expect:成功-----"
        self.logger.info(text_2)
        restore_cmd_1 = f'''
            if [[ -e {self.postgresql_path} && -e {self.postgresql_bak_path} ]]; 
            then rm -rf {self.postgresql_path}; 
            mv {self.postgresql_bak_path} {self.postgresql_path}; 
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_1)
        restore_primary_1 = self.primary_node.sh(restore_cmd_1).result()
        restore_standby_1 = self.standby_node.sh(restore_cmd_1).result()
        self.logger.info(restore_primary_1)
        self.logger.info(restore_standby_1)
        restore_cmd_2 = f'''
            if [[ -e {self.pg_hba_path} && -e {self.pg_hba_bak_path} ]];
            then rm -rf {self.pg_hba_path};  
            mv {self.pg_hba_bak_path} {self.pg_hba_path};
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_2)
        restore_primary_2 = self.primary_node.sh(restore_cmd_2).result()
        restore_standby_2 = self.standby_node.sh(restore_cmd_2).result()
        self.logger.info(restore_primary_2)
        self.logger.info(restore_standby_2)

        text_3 = "-----step11.3:删除文件，expect:成功-----"
        self.logger.info(text_3)
        del_cmd = f'rm -rf {self.dir_replication};'
        self.logger.info(del_cmd)
        self.primary_node.sh(del_cmd)
        self.standby_node.sh(del_cmd)
        file_cmd = f'''if [ -d {self.dir_replication} ]; 
            then echo "exists"; else echo "not exists"; fi'''
        primy_file_msg = self.primary_node.sh(file_cmd).result()
        standy_file_msg = self.standby_node.sh(file_cmd).result()
        self.logger.info(primy_file_msg)
        self.logger.info(standy_file_msg)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_msg, '执行失败' + text_1)
        self.assertEqual('OK', restore_primary_1, '执行失败' + text_2)
        self.assertEqual('OK', restore_standby_1, '执行失败' + text_2)
        self.assertEqual('OK', restore_primary_2, '执行失败' + text_2)
        self.assertEqual('OK', restore_standby_2, '执行失败' + text_2)
        self.assertEqual("not exists", primy_file_msg, '执行失败' + text_3)
        self.assertEqual("not exists", standy_file_msg, '执行失败' + text_3)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")